package io.kimmking.common.rocketmq.repository;

import com.google.common.base.Preconditions;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.roaringbitmap.RoaringBitmap;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.transaction.annotation.Transactional;

/**
 * 在发送客户端发送事务性消息并且实现回查Listener
 * 使用方法：
 * 1、使用时需要extends这个类，且加上注解@RocketMQTransactionListener
 * 2、同时实现localTransactionExecutor()方法
 */
@Slf4j
//@RocketMQTransactionListener
public abstract class DistrbuitedTxProducerWrapper implements RocketMQLocalTransactionListener {
    //用于存放事务ID(也可以存储在数据库中)
    private final RoaringBitmap bm = new RoaringBitmap();

    /**
     * 执行本地事务(业务)方法
     */
    @Transactional(rollbackFor = RuntimeException.class)
    protected abstract boolean localTransactionExecutor(Object... o);

    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        log.info("本地事务开始执行。消息体对象数据：{}，arg参数对象数据：{}", message, o);
        RocketMQLocalTransactionState rlts = RocketMQLocalTransactionState.UNKNOWN;
        MessageHeaders headers = message.getHeaders();
        // 获取事务ID
        String transactionId = headers.get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID, String.class);
        if(transactionId==null) {
            log.error("lack of transactionId in headers of message!");
            return RocketMQLocalTransactionState.ROLLBACK;
        }
        // 消息体内容
//      String body = new String((byte[]) message.getPayload());

        try {
            // TODO 调用发券业务逻辑，发券业务逻辑理论上应该是事务方法 @Transactional，若失败则进行发券回滚，此处拦截也进行拦截
            boolean re = localTransactionExecutor(o);
            if(!re) {
                rlts = RocketMQLocalTransactionState.ROLLBACK;
            } else{
                rlts = RocketMQLocalTransactionState.COMMIT;
                bm.add(Integer.parseInt(Preconditions.checkNotNull(transactionId)));
            }

            // TODO 如果发券成功，在同一事务中建议将transaction_id保存一下，可以保存到数据库
            // 此处我们创建个表来存储事务消息，方便checkLocalTransaction回查
            // 表结构类似如下：
            // CREATE TABLE `d_test_weiif`.`无标题`  (
            //   `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键id',
            //   `transaction_id` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT 'mq消息事务id(消息id)',
            //   `topic` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT 'mq消息主题',
            //   `keys` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT 'mq消息key',
            //   `body` text CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL COMMENT '消息体内容，json格式',
            //   `add_time` datetime(0) NOT NULL DEFAULT CURRENT_TIMESTAMP(0) COMMENT '添加时间',
            //   `add_user_id` bigint(20) NULL DEFAULT NULL COMMENT '添加人',
            //    PRIMARY KEY (`id`) USING BTREE
            // ) ENGINE = InnoDB AUTO_INCREMENT = 6 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci COMMENT = 'RocketMQ 事务消息日志表' ROW_FORMAT = Compact;

            //其实只需要transaction_id即可，如果回查时能根据transaction_id回查到，则证明该消息正常，执行commit操作
        } catch (Exception e) {
            // 出现异常，MQ消息进行回滚
            log.error("发券出现异常，事务回滚，MQ消息回滚，异常信息：{}", e.getMessage());
            rlts = RocketMQLocalTransactionState.ROLLBACK;
        }

        return rlts;
    }

    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        MessageHeaders headers = message.getHeaders();
        // 获取事务ID
        String transactionId = (String) headers.get(RocketMQHeaders.PREFIX + RocketMQHeaders.TRANSACTION_ID);
        log.info("RocketMQ发起回查，回查事务消息ID：{}", transactionId);
        log.info("RocketMQ发起回查，回查事务消息数据：{}", message);
        try {
            // TODO 回查事务ID是否存在，此处是在RoaringBitmap中查找
            if (bm.contains(Integer.parseInt(Preconditions.checkNotNull(transactionId)))) {
                log.info("回查事务ID确认成功，开始commit消息");
                return RocketMQLocalTransactionState.COMMIT;
            } else {
                log.info("回查事务ID失败，MQ消息事务回滚");
                return RocketMQLocalTransactionState.ROLLBACK;
                // 如果此处设置成 RocketMQLocalTransactionState.UNKNOWN，则RocketMQ还会继续回查
            }
        } catch (Exception e) {
            // 查询期间出现异常，则继续下次回查
            log.error("RocketMQ发起回查，回查事务消息ID：{}，回查出现异常，异常信息：{}", transactionId, e.getMessage());
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
}
